package com.atguigu.flink.chapter09_cep;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2022/10/29
 *
 *      套路:
 *              ①有一个流，在流上设置水印(基于事件时间)，ProcessingTime不支持
 *                      基于数据的事件时间的顺序处理。
 *              ②设置一个Pattern(模式)
 *              ③调用CEP的库，把pattern作用到流上，得到 模式流(存放匹配到的数据)
 *              ④从模式流中获取匹配的数据
 *
 */
public class Demo1_HelloWorld
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((element, recordTimestamp) -> element.getTs());

        //①有一个流，在流上设置水印(基于事件时间)，ProcessingTime不支持
        SingleOutputStreamOperator<WaterSensor> ds = env
            .readTextFile("data/cep.txt")
            .map(new MapFunction<String, WaterSensor>()
            {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(
                        data[0],
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2])
                    );
                }
            })
            .assignTimestampsAndWatermarks(watermarkStrategy);

        /*
                ②定义模式
                    每一个模式必须有一个名字(string)，随便写。
         */
        Pattern<WaterSensor, WaterSensor> pattern = Pattern
                                          .<WaterSensor>begin("匹配s1")   //规则名为  匹配s1
                                          .where(new SimpleCondition<WaterSensor>()
                                          {
                                              //定义过滤规则
                                              @Override
                                              public boolean filter(WaterSensor value) throws Exception {
                                                  return "s1".equals(value.getId());
                                              }
                                          })
                                            //.times(2)   //匹配s1 定义的规则 会应用2次
                                            //.times(2,3)   //匹配s1 定义的规则 会应用2-3次
                                            //.oneOrMore()    //匹配s1 定义的规则 会应用1次-----N次
                                               .timesOrMore(2)  //匹配s1 定义的规则 会应用2次-----N次
                                            .until(new SimpleCondition<WaterSensor>()  //设置多次数匹配规则的终止条件
                                            {
                                                @Override
                                                public boolean filter(WaterSensor value) throws Exception {
                                                    return "s3".equals(value.getId());
                                                }
                                            })
                                            ;

        //③调用CEP的库，把pattern作用到流上，得到 模式流(存放匹配到的数据)
        PatternStream<WaterSensor> ds2 = CEP.pattern(ds, pattern);

         //④从模式流中获取匹配的数据
        ds2.select(new PatternSelectFunction<WaterSensor, String>()
        {
            /*
                    挑选匹配到的数据

                    Map<String, List<WaterSensor>> map：
                        key： 规则名
                        value: 匹配到的数据

                     定义的pattern是对每一条数据进行匹配，每一次匹配的结果都会传递到map
             */
            @Override
            public String select(Map<String, List<WaterSensor>> map) throws Exception {
                return map.toString();
            }
        }).print().setParallelism(1);


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
